package rabbitmq

import (
	"fmt"
	amqp "github.com/rabbitmq/amqp091-go"
	"sync"
)

// ExchangeType 枚举交换机类型
type ExchangeType string

const (
	DirectExchange  ExchangeType = "direct"
	FanoutExchange  ExchangeType = "fanout"
	TopicExchange   ExchangeType = "topic"
	HeadersExchange ExchangeType = "headers"
)

// ConnectionPool 用于管理 RabbitMQ 连接池
type ConnectionPool struct {
	mu          sync.Mutex
	connections []*amqp.Connection // 存储连接的切片
	maxSize     int                // 最大连接数
}

// 创建新的连接池
func NewConnectionPool(maxSize int) *ConnectionPool {
	return &ConnectionPool{
		maxSize:     maxSize,
		connections: make([]*amqp.Connection, 0, maxSize),
	}
}

// GetConnection 获取一个连接，如果没有空闲连接则创建新的连接
func (pool *ConnectionPool) GetConnection() (*amqp.Connection, error) {
	pool.mu.Lock()
	defer pool.mu.Unlock()

	// 如果池中有空闲连接，则返回
	if len(pool.connections) > 0 {
		conn := pool.connections[len(pool.connections)-1]
		pool.connections = pool.connections[:len(pool.connections)-1]
		return conn, nil
	}

	// 如果池中没有空闲连接且未超出最大连接数，创建新连接
	if len(pool.connections) < pool.maxSize {
		conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
		if err != nil {
			return nil, err
		}
		return conn, nil
	}

	// 如果连接池已满，返回错误
	return nil, fmt.Errorf("connection pool limit reached")
}

// 归还连接到连接池
func (pool *ConnectionPool) ReleaseConnection(conn *amqp.Connection) {
	pool.mu.Lock()
	defer pool.mu.Unlock()
	// 如果池还没有满，将连接放回池中
	if len(pool.connections) < pool.maxSize {
		pool.connections = append(pool.connections, conn)
	} else {
		// 如果池已满，关闭连接
		conn.Close()
	}
}

// Producer 推送消息的工具类
type Producer struct {
	pool *ConnectionPool
}

// 创建新的 Producer
func NewProducer(pool *ConnectionPool) *Producer {
	return &Producer{
		pool: pool,
	}
}
func (p *Producer) Push(exchange string, exchangeType ExchangeType, routingKey, message string) error {
	conn, err := p.pool.GetConnection()
	if err != nil {
		return fmt.Errorf("failed to get connection: %v", err)
	}
	defer p.pool.ReleaseConnection(conn)

	ch, err := conn.Channel()
	if err != nil {
		return fmt.Errorf("failed to create channel: %v", err)
	}
	defer ch.Close()

	// 声明交换机（如果交换机已存在可省略这一步）
	err = ch.ExchangeDeclare(
		exchange,
		string(exchangeType),
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		return fmt.Errorf("failed to declare exchange: %v", err)
	}

	// 👇 新增：声明队列（名称可根据 routingKey 衍生）
	queueName := fmt.Sprintf("queue.%s", routingKey)
	_, err = ch.QueueDeclare(
		queueName, // 队列名称（由 routingKey 推导）
		true,      // durable，与交换机一致
		true,      // auto-delete
		false,     // exclusive
		false,     // no-wait
		nil,       // args
	)
	if err != nil {
		return fmt.Errorf("failed to declare queue: %v", err)
	}

	// 👇 新增：绑定队列和交换机
	err = ch.QueueBind(
		queueName, // queue name
		routingKey,
		exchange,
		false,
		nil,
	)
	if err != nil {
		return fmt.Errorf("failed to bind queue: %v", err)
	}

	// 发布消息
	err = ch.Publish(
		exchange,
		routingKey,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		},
	)
	if err != nil {
		return fmt.Errorf("failed to publish message: %v", err)
	}

	fmt.Printf("✅ Sent message to exchange '%s' with routing key '%s': %s\n", exchange, routingKey, message)
	return nil
}

// Consumer 订阅消息的工具类
type Consumer struct {
	pool *ConnectionPool
}

// 创建新的 Consumer
func NewConsumer(pool *ConnectionPool) *Consumer {
	return &Consumer{
		pool: pool,
	}
}

func (c *Consumer) Subscribe(
	exchange string,
	exchangeType ExchangeType,
	routingKeys []string, // 支持多个 routingKey
	queueName string,
	handler func(msg string),
) error {
	conn, err := c.pool.GetConnection()
	if err != nil {
		return err
	}
	defer c.pool.ReleaseConnection(conn)
	ch, err := conn.Channel()
	if err != nil {
		return err
	}
	defer ch.Close()

	// 声明交换机
	err = ch.ExchangeDeclare(exchange, string(exchangeType), true, false, false, false, nil)
	if err != nil {
		return err
	}

	// 声明持久队列
	_, err = ch.QueueDeclare(queueName, true, false, false, false, nil)
	if err != nil {
		return err
	}

	// 绑定多个 routingKey
	for _, key := range routingKeys {
		err = ch.QueueBind(queueName, key, exchange, false, nil)
		if err != nil {
			return err
		}
	}

	// 开始消费
	msgs, err := ch.Consume(queueName, "", true, false, false, false, nil)
	if err != nil {
		return err
	}

	for msg := range msgs {
		handler(string(msg.Body))
	}
	return nil
}
